package com.caption.mq.core;

import com.caption.mq.entity.MqSubscribeDO;
import com.caption.mq.listener.AbstractMqListener;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class DefaultRocketMqConsumerManager implements MqConsumerManger {
    private static Map<String, AbstractMqConsumer> consumers = new ConcurrentHashMap<>(256);
    //TODO 注入工厂
    private RocketMqFactory rocketMqFactory;

    @Override
    public void stopConsumer(MqSubscribeDO mqSubscribeDO) {
        String consumerId = mqSubscribeDO.getConsumerId();
        if (consumerId != null) {
            consumers.get(consumerId).shutdown();
            consumers.remove(consumerId);
        }
    }
    @Override
    public void startConsumer(AbstractMqConsumer consumer) {
        if (consumer != null) {
            consumer.start();
        }
    }
    @Override
    public AbstractMqConsumer createConsumer(MqSubscribeDO mqSubscribeDO) {
        String consumerId = mqSubscribeDO.getConsumerId();
        if (consumers.get(consumerId) != null) {
            return consumers.get(consumerId);
        }
        try {
            AbstractMqConsumer consumer = rocketMqFactory.getConsumerInstance(mqSubscribeDO);
            consumer.subscribe(mqSubscribeDO.getTopic(), mqSubscribeDO.getTag());
            AbstractMqListener mqListener = rocketMqFactory.getMqListenerInstance(mqSubscribeDO);
            consumer.registerMessageListener(mqListener);
            consumers.put(consumerId, consumer);
            return consumer;
        } catch (Exception e) {
            throw e;
        }
    }
    @Override
    public void stopAllConsumer() {
        for (String consumerId : consumers.keySet()) {
            consumers.get(consumerId).shutdown();

        }
        consumers.clear();
    }


}
